package com.bw.week2

import org.apache.spark.sql.SparkSession

object One {
  def main(args: Array[String]): Unit = {
//    1.初始化：实例化SparkContext上下文对象，设置AppName和Master；（3分）
val spark = SparkSession.builder().master("local[*]").appName("app").getOrCreate()
    import spark.implicits._
    val sc = spark.sparkContext
//    2.数据加载：读取交通数据，封装RDD集合；（4分）
val rdd = sc.textFile("data/traffic-data.txt")
    val rdd2 = rdd.filter(line => {
      val strings = line.split(",")
      strings.length == 7
    }).map(line => {
      val str = line.split(",")
      jiao(str(0).toInt, str(1).toInt, str(2), str(3), str(4).toInt, str(5).toInt, str(6).toInt)
    })
    //    3.数据加载：统计交通数据条目数和显示前10天样本数据；（3分)
    println(rdd2.count())
    rdd2.take(10).foreach(println)
//    4.分析1：基于Scala代码编写，调用RDD算子，获取车速大于90以上交通数据；（5分）
val r4 = rdd2.filter(a => a.speed > 90)
    r4.foreach(println)
//    5.分析2：基于Scala代码编写，调用RDD算子，计算各个区域交通流量；（5分）
val r5 = rdd2.groupBy(q => q.qid).map(a => {
  (a._1, a._2.size)
})
    r5.foreach(println)
//    6.分析2：对第5题计算结果，按照交通流量降序排序，获取Top3流量区域；（5分）
val r6 = r5.sortBy(a => a._2, false).take(3)
    r6.foreach(println)
//    7.分析3：基于Scala代码编写，调用RDD算子，计算各个省份车牌的交通流量；（5分）
val r7 = rdd2.groupBy(s => s.cid.split("-")(0)).map(c => (c._1, c._2.size))
    r7.foreach(println)
//    8.分析3：对第7题计算结果，保存HDFS文件系统，并查看文件数据；（5分）
//    r7.saveAsTextFile("hdfs://node00:8020/day/r7")
//    9.分析3：对第7题计算结果，保存MySQL数据库表，实现覆盖模式，并查看表中数据；（5分）
//val r9 = r7.toDF()
//    r9.write
//      .format("jdbc")
//      .mode("overwrite")
//      .option("url", "jdbc:mysql://node00:3306/test?useSSL=false")
//      .option("dbtable", "test.r9")
//      .option("user", "root")
//      .option("password", "root")
//      .save()

    sc.stop()
  }
}

case class jiao(var jcid:BigInt,var jkid:BigInt,var cid:String,var sdate:String,var speed:BigInt,var did:BigInt,var qid:BigInt)
